package com.bw.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TradeSkuOrderBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.function.DimAsyncFunction;
import com.bw.gmall.realtime.common.function.DorisMapFunction;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import com.bw.gmall.realtime.common.util.HBaseUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.Iterator;
import java.util.concurrent.TimeUnit;

public class DwsTradeSkuOrderAsyncCacheRealWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeSkuOrderAsyncCacheRealWindow().start(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL,Constant.DWS_TRADE_SKU_ORDER_WINDOW,4,10095);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {


        //etl
//        dataStreamSource.print();
        SingleOutputStreamOperator<JSONObject> streamOperator = EltStream(dataStreamSource);

        //水位线
        SingleOutputStreamOperator<JSONObject> waterStream = WaterStream(streamOperator);

        // 2. 按照 order_detail_id // 4.防止下游重复计算
        SingleOutputStreamOperator<TradeSkuOrderBean> process = getStreamOperator(waterStream);

        // 5.以sku_id分组
        KeyedStream<TradeSkuOrderBean, String> keyedStream = process.keyBy(x -> x.getSkuId());


        //开窗 聚合
        SingleOutputStreamOperator<TradeSkuOrderBean> reduce = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean t1, TradeSkuOrderBean t2) throws Exception {

                        t1.setActivityReduceAmount(t1.getActivityReduceAmount().add(t2.getActivityReduceAmount()));
                        t1.setCouponReduceAmount(t1.getCouponReduceAmount().add(t2.getCouponReduceAmount()));
                        t1.setOrderAmount(t1.getOrderAmount().add(t2.getOrderAmount()));
                        t1.setOriginalAmount(t1.getOriginalAmount().add(t2.getOriginalAmount()));

                        return t1;
                    }
                }, new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<TradeSkuOrderBean> iterable, Collector<TradeSkuOrderBean> collector) throws Exception {

                        long start = timeWindow.getStart();
                        long end = timeWindow.getEnd();

                        Iterator<TradeSkuOrderBean> iterator = iterable.iterator();
                        TradeSkuOrderBean tradeSkuOrderBean = iterator.next();

                        tradeSkuOrderBean.setStt(DateFormatUtil.tsToDateTime(start));
                        tradeSkuOrderBean.setEdt(DateFormatUtil.tsToDateTime(end));
                        tradeSkuOrderBean.setCurDate(DateFormatUtil.tsToDateTime(System.currentTimeMillis()));
                        collector.collect(tradeSkuOrderBean);
                    }
                });


        // 4. join 维度
//        reduce.print();

        //sku
        SingleOutputStreamOperator<TradeSkuOrderBean> skuinfo_Wait1= AsyncDataStream.unorderedWait(reduce, new DimAsyncFunction<TradeSkuOrderBean>("dim_sku_info") {
            @Override
            public String getDimId(TradeSkuOrderBean o) {
                return o.getSkuId();
            }

            @Override
            public void setDim(TradeSkuOrderBean o, JSONObject dimInfo) {
                o.setSkuName(dimInfo.getString("sku_name"));
                o.setSpuId(dimInfo.getString("spu_id"));
                o.setCategory3Id(dimInfo.getString("category3_id"));
                o.setTrademarkId(dimInfo.getString("tm_id"));
            }
        }, 100, TimeUnit.SECONDS);
//        skuinfo_Wait1.print();

        //SPU
        SingleOutputStreamOperator<TradeSkuOrderBean> spuinfo_Wait2= AsyncDataStream.unorderedWait(skuinfo_Wait1, new DimAsyncFunction<TradeSkuOrderBean>("dim_spu_info") {
            @Override
            public String getDimId(TradeSkuOrderBean o) {
                return o.getSpuId();
            }

            @Override
            public void setDim(TradeSkuOrderBean o, JSONObject dimInfo) {
                o.setSpuName(dimInfo.getString("spu_name"));
            }
        }, 1000, TimeUnit.SECONDS);

//        spuinfo_Wait2.print();


        //tm_id
        SingleOutputStreamOperator<TradeSkuOrderBean> tm_Wait2= AsyncDataStream.unorderedWait(spuinfo_Wait2, new DimAsyncFunction<TradeSkuOrderBean>("dim_base_trademark") {
            @Override
            public String getDimId(TradeSkuOrderBean o) {
                return o.getTrademarkId();
            }

            @Override
            public void setDim(TradeSkuOrderBean o, JSONObject dimInfo) {
                o.setTrademarkName(dimInfo.getString("tm_name"));
            }
        }, 100, TimeUnit.SECONDS);


        //cate3
        SingleOutputStreamOperator<TradeSkuOrderBean> cate3_Wait2= AsyncDataStream.unorderedWait(tm_Wait2, new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category3") {
            @Override
            public String getDimId(TradeSkuOrderBean o) {
                return o.getCategory3Id();
            }

            @Override
            public void setDim(TradeSkuOrderBean o, JSONObject dimInfo) {
                o.setCategory3Name(dimInfo.getString("name"));
                o.setCategory2Id(dimInfo.getString("category2_id"));
            }
        }, 100, TimeUnit.SECONDS);


        //cate2
        SingleOutputStreamOperator<TradeSkuOrderBean> cate2_Wait2= AsyncDataStream.unorderedWait(cate3_Wait2, new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category2") {
            @Override
            public String getDimId(TradeSkuOrderBean o) {
                return o.getCategory2Id();
            }

            @Override
            public void setDim(TradeSkuOrderBean o, JSONObject dimInfo) {
                o.setCategory2Name(dimInfo.getString("name"));
                o.setCategory1Id(dimInfo.getString("category1_id"));
            }
        }, 100, TimeUnit.SECONDS);


        //cate1
        SingleOutputStreamOperator<TradeSkuOrderBean> cate1_Wait2= AsyncDataStream.unorderedWait(cate2_Wait2, new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category1") {
            @Override
            public String getDimId(TradeSkuOrderBean o) {
                return o.getCategory1Id();
            }

            @Override
            public void setDim(TradeSkuOrderBean o, JSONObject dimInfo) {
                o.setCategory1Name(dimInfo.getString("name"));
            }
        }, 100, TimeUnit.SECONDS);




        cate1_Wait2.print();

        //写入 doris
//        mapstream.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSinks(Constant.DWS_TRADE_SKU_ORDER_WINDOW));


    }

    private static SingleOutputStreamOperator<TradeSkuOrderBean> getStreamOperator(SingleOutputStreamOperator<JSONObject> waterStream) {
         return waterStream.keyBy(x -> x.getString("id")).process(new ProcessFunction<JSONObject, TradeSkuOrderBean>() {

            //处理 回撤流
            MapState<String, BigDecimal> state;


            @Override
            public void open(Configuration parameters) {
                MapStateDescriptor<String, BigDecimal> stateDescriptor = new MapStateDescriptor<>("state", String.class, BigDecimal.class);
                state = getRuntimeContext().getMapState(stateDescriptor);


            }

            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, TradeSkuOrderBean>.Context context, Collector<TradeSkuOrderBean> collector) throws Exception {

                //逻辑 ： 来到的数据 先存入 状态中，然后下一次来的数据 减去 上次的值 在存入 状态， 状态里相加就是 没有回撤流的结果

                //取状态的值 若 状态为空 设为 0   模拟从第一次 取状态的值
                BigDecimal originalAmount = state.get("split_original_amount");
                BigDecimal activityReduceAmount = state.get("splitActivityAmount");
                BigDecimal couponReduceAmount = state.get("splitCouponAmount");
                BigDecimal totalorderAmount = state.get("splitTotalAmount");

                //避免为空
                totalorderAmount = totalorderAmount == null ? new BigDecimal(0) : totalorderAmount;
                originalAmount = originalAmount == null ? new BigDecimal(0) : originalAmount;
                activityReduceAmount = activityReduceAmount == null ? new BigDecimal(0) : activityReduceAmount;
                couponReduceAmount = couponReduceAmount == null ? new BigDecimal(0) : couponReduceAmount;


                //获取 数据
                Long ts = jsonObject.getLong("ts");

                String skuId = jsonObject.getString("sku_id");
                String id = jsonObject.getString("id");

                String skuName = jsonObject.getString("sku_name");
                BigDecimal splitTotalAmount = jsonObject.getBigDecimal("split_total_amount");
                BigDecimal splitActivityAmount = jsonObject.getBigDecimal("split_activity_amount");
                BigDecimal splitCouponAmount = jsonObject.getBigDecimal("split_coupon_amount");

                BigDecimal sku_num = jsonObject.getBigDecimal("sku_num");
                BigDecimal orderPrice = jsonObject.getBigDecimal("order_price");
                //乘法
                BigDecimal split_original_amount = orderPrice.multiply(sku_num);



                // 获取 状态内的数据 进行 去除 回撤流 并 返回流
                collector.collect(TradeSkuOrderBean.builder()
                        .skuId(skuId)
                        .orderAmount(splitTotalAmount.subtract(totalorderAmount))
                        .activityReduceAmount(splitActivityAmount.subtract(activityReduceAmount))
                        .couponReduceAmount(splitCouponAmount.subtract(couponReduceAmount))
                        .skuName(skuName)
                        .originalAmount(split_original_amount.subtract(originalAmount))
                        .curDate(DateFormatUtil.tsToDateTime(ts))
                                .ts(ts)

                        .orderDetailId(id)
                        .build());

                //将本次数据 提交到 状态中
                state.put("split_original_amount", split_original_amount);
                state.put("splitActivityAmount", splitActivityAmount);
                state.put("splitCouponAmount", splitCouponAmount);
                state.put("splitTotalAmount", splitTotalAmount);
            }
        });

    }

    private static  SingleOutputStreamOperator<JSONObject> WaterStream(SingleOutputStreamOperator<JSONObject> streamOperator) {
        return streamOperator.assignTimestampsAndWatermarks(WatermarkStrategy
                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner((event, timestamp) -> event.getLong("ts")).withIdleness(Duration.ofSeconds(1)));
    }

    private static SingleOutputStreamOperator<JSONObject> EltStream(DataStreamSource<String> dataStreamSource) {
         return dataStreamSource.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {

                if (s != null) {

                    JSONObject jsonObject = JSON.parseObject(s);

                    if (jsonObject != null) {

                        String skuId = jsonObject.getString("sku_id");
                        Long ts = jsonObject.getLong("ts");
                        if (skuId != null && ts != null) {

                            jsonObject.put("ts", ts * 1000);

                            collector.collect(jsonObject);
                        }
                    }
                }
            }
        });
    }
}
